import time
import queue
import logging
import importlib
import threading
from types import FunctionType
from typing import List, AnyStr
from datetime import datetime

from .lock import LocalLock, LockClass
from .stores.job import Job
from .stores.base import BaseJobStore, JobLookupError, ConflictingIdError


class JobContinueException(Exception):
    '''继续调度任务'''


class JobStopException(Exception):
    '''停止调度任务'''


job_queue = queue.Queue(maxsize=1000)


class Controller(object):

    def __init__(self,
                 max_workers=50,
                 min_avai_workers_threshold=5,
                 per_job_running_duration=30,
                 max_wait_seconds=30,
                 store: BaseJobStore = None,
                 lockClass: LockClass = LocalLock,
                 logger: logging.Logger = None):
        '''
        :param int max_workers: 工作线程数
        :param int min_avai_workers_threshold: 拉取任务时空闲工作线程数阈值
        :param int per_job_running_duration: 当前控制器锁定任务的时长
        :param BaseJobStore store: 存储引擎
        :param LockClass lockClass: 本地锁-单实例控制器，分布式锁-多实例控制器
        :param logging.Logger logger: 日志记录器
        :param int max_wait_seconds: 没有任务的时候，主线程每隔`max_wait_seconds`秒主动拉取一次任务
        '''
        assert issubclass(store.__class__, BaseJobStore)
        assert isinstance(store, BaseJobStore)
        assert issubclass(lockClass, LockClass)
        self.max_workers = max_workers
        self.min_avai_workers_threshold = min_avai_workers_threshold
        self.per_job_runing_duration = per_job_running_duration
        self._max_wait_seconds = max_wait_seconds
        self.store = store
        self._avai_worker_cnt = max_workers
        self._event = threading.Event()
        self._fetch_jobs_lock = lockClass('fetch_jobs')
        self._worker_cnt_lock = threading.Lock()
        self._handlers = {}
        self._job_queue = queue.Queue(maxsize=max_workers)
        self._terminated_job_cnt = 0
        self._local_jobs = {}
        self._local_jobs_lock = threading.Lock()
        if logger:
            self._logger = logger
        else:
            fmt = logging.Formatter("%(asctime)s %(name)s %(levelname)s [%(filename)s:%(lineno)d:%(funcName)s] %(message)s")
            self._logger = logging.getLogger('python-jobs')
            self._logger.setLevel('DEBUG')
            hd = logging.StreamHandler()
            hd.setLevel('DEBUG')
            hd.setFormatter(fmt)
            self._logger.addHandler(hd)

    def _add_local_job(self, job_id: AnyStr):
        with self._local_jobs_lock:
            self._local_jobs[job_id] = job_id

    def _remove_local_job(self, job_id: AnyStr):
        with self._local_jobs_lock:
            self._local_jobs.pop(job_id, '')

    def add_job(self, job: Job):
        try:
            self.store.add_job(job)
        except ConflictingIdError:
            return
        except AssertionError as err:
            self._logger.error('{}'.format(err), exc_info=True)
        self.wakeup()

    def update_job(self, job: Job):
        try:
            self.store.update_job(job)
        except JobLookupError as err:
            self._logger.error('{}'.format(err), exc_info=True)
        finally:
            self.wakeup()

    def remove_job(self, job: Job):
        self.store.remove_job(job)
        self.wakeup()

    def _process_jobs(self):
        self._fetch_jobs_lock.acquire()
        try:
            self._logger.debug('当前空闲worker数量为{}, 尝试拉取{}个任务'.format(self._avai_worker_cnt, self._avai_worker_cnt))
            jobs: List[Job] = self.store.get_due_jobs(int(time.time()), self._avai_worker_cnt, list(self._local_jobs.keys()))
            if not jobs:
                return
            self.store.lock_jobs(jobs, int(time.time()) + self.per_job_runing_duration)
            for job in jobs:
                self._add_local_job(job.job_id)
                self._job_queue.put(job)
        except Exception as err:
            self._logger.error('处理任务失败: {}'.format(err), exc_info=True)
        finally:
            self._fetch_jobs_lock.release()

    def _worker(self):
        while True:
            job: Job = self._job_queue.get()
            if job.delete_time > 0:
                self._logger.warning('job({})被重复调度'.format(job.id))
                continue
            with self._worker_cnt_lock:
                self._avai_worker_cnt -= 1
            job.sched_times += 1
            max_sched_times = job.max_sched_times
            start_at = time.time()
            to_be_continue = True
            args = ()
            kwargs = {}
            try:
                args = job.func_args
                kwargs = job.func_kwargs
                max_sched_times = kwargs.get('max_sched_times', max_sched_times)
            except Exception as err:
                self._logger.error(f'任务{job.id}加载参数报错: {err}', exc_info=True)
                self.remove_job(job)
                continue
            to_be_continue = False if job.sched_times >= max_sched_times and max_sched_times > 0 else True
            try:
                f = self._handlers[job.func_name]
                f(*args, **kwargs)
                job.delete_time = int(time.time())
            except JobStopException:
                job.delete_time = int(time.time())
            except JobContinueException:
                if to_be_continue:
                    self._logger.error('处理任务{}失败'.format(job.id))
                    job.next_run_time = int(time.time()) + job.sched_times * job.delay_seconds
                else:
                    job.delete_time = int(time.time())
                    self._logger.error('任务{}达到最大处理次数限制'.format(job.id))
                if job.sched_times / max_sched_times >= 0.5:
                    self._logger.error('任务{}重试次数已经达到{}'.format(job.id, job.sched_times))
            except Exception as err:
                if to_be_continue:
                    self._logger.error('处理任务{}失败: {}'.format(job.id, err), exc_info=True)
                    job.next_run_time = int(time.time()) + job.sched_times * job.delay_seconds
                else:
                    job.delete_time = int(time.time())
                    self._logger.error('任务{}达到最大处理次数限制: {}'.format(job.id, err))
                if job.sched_times / max_sched_times >= 0.5:
                    self._logger.error('任务{}重试次数已经达到{}'.format(job.id, job.sched_times))
            finally:
                end_at = time.time()
                cost = round(end_at - start_at, 2)
                next_run_time = None if job.delete_time > 0 else datetime.fromtimestamp(job.next_run_time)
                delay_seconds = None if job.delete_time > 0 else job.sched_times + job.delay_seconds
                if delay_seconds:
                    self._logger.info('任务({})已经执行({})次, 下次运行时间: {} ({}秒后) -- 耗时{}s'.format(job.id, job.sched_times, next_run_time, delay_seconds, cost))
                else:
                    self._logger.info('任务({})结束调度'.format(job.id))
                try:
                    self.update_job(job)
                    self._remove_local_job(job.job_id)
                except Exception as err:
                    self._logger.error('更新任务({})失败: {}'.format(job.job_id, err), exc_info=True)
                with self._worker_cnt_lock:
                    if job.delete_time > 0:
                        job.next_run_time = None
                        self._terminated_job_cnt += 1
                    self._avai_worker_cnt += 1

    def _launch_wokers(self):
        for _ in range(self.max_workers):
            th = threading.Thread(target=self._worker, daemon=True)
            th.start()

    def wakeup(self):
        self._event.set()

    def start(self, background=True):
        self._launch_wokers()
        th = threading.Thread(target=self._main_loop, name='main_loop', daemon=True)
        th.start()
        th2 = threading.Thread(target=self._add_job_consumer, name='add_job_consumer', daemon=True)
        th2.start()
        if not background:
            th.join()

    def _get_wait_seconds(self):
        next_run_time = self.store.get_next_run_time()
        if next_run_time is None:
            return self._max_wait_seconds
        wait_seconds = next_run_time - int(time.time())
        if wait_seconds < 0:
            wait_seconds = 0
        if wait_seconds > self._max_wait_seconds:
            return self._max_wait_seconds
        return wait_seconds

    def _main_loop(self):
        wait_seconds = 0
        while True:
            self._event.wait(wait_seconds)
            self._event.clear()
            # 当前空闲工作线程数未达到拉取任务的阈值不进行处理
            if self._avai_worker_cnt < self.min_avai_workers_threshold:
                self._logger.debug('当前空闲工作线程数未达到拉取任务的阈值, 跳过当前处理请求')
                wait_seconds = 1
            else:
                self._process_jobs()
                wait_seconds = self._get_wait_seconds()
            self._logger.debug('调度器({})秒后再次唤醒, 终止任务数{}'.format(wait_seconds, self._terminated_job_cnt))

    def new_job(self,
                job_id: AnyStr,
                func_name: AnyStr,
                max_sched_times: int = 200,
                delay_seconds: int = 10,
                func_args: list = [],
                func_kwargs: dict = {}) -> Job:
        '''创建新任务'''
        job = Job()
        job.job_id = job.id = job_id
        job.func_name = func_name
        job.func_args = func_args
        job.func_kwargs = func_kwargs
        job.max_sched_times = max_sched_times
        job.delay_seconds = delay_seconds
        self.store.add_job(job)

    def get_job(self, job_id: AnyStr) -> Job:
        return self.store.lookup_job(job_id=job_id)

    def register(self, func: FunctionType):
        self._handlers[func.__name__] = func
        return func

    def autodiscover_jobs(self,
                          packages: list = None,
                          related_name='jobs'):
        """Auto-discover jobs modules.
        Arguments:
            packages (List[str]): List of packages to search.
                This argument may also be a callable, in which case the
                value returned is used (for lazy evaluation).
            related_name (str): The name of the module to find.  Defaults
                to "controller_jobs": meaning "look for 'module.controller_jobs' for every
                module in ``packages``.".  If ``None`` will only try to import
                the package, i.e. "look for 'module'".
            force (bool): By default this call is lazy so that the actual
                auto-discovery won't happen until an application imports
                the default modules.  Forcing will cause the auto-discovery
                to happen immediately..
        """
        job_modules = set()
        for package in packages:
            job_module = self._autodiscover_jobs(package, related_name)
            if job_module:
                self._logger.info("Import package: {}".format(job_module.__name__))
                job_modules.add(job_module.__name__)
        return job_modules

    def _autodiscover_jobs(self, package, related_name="jobs"):
        module_name = '{0}.{1}'.format(package, related_name)

        try:
            importlib
            return importlib.import_module(module_name)
        except ImportError as e:
            import_exc_name = getattr(e, 'name', module_name)
            if import_exc_name is not None and import_exc_name != module_name:
                raise e
            return

    def _add_job_consumer(self):
        while True:
            job: Job = job_queue.get()
            try:
                self.add_job(job)
            except Exception as err:
                self._logger.error(f'处理新增任务{job}出错: {err}', exc_info=True)
